package com.nami.logic.disruptor.handler;

import com.lmax.disruptor.EventHandler;
import com.nami.logic.disruptor.event.TaskEvent;
import com.nami.logic.utils.ConverTaskUtil;
import com.nami.store.api.request.conversation.task.ConverTaskAddReq;
import com.nami.store.api.service.RemoteConverTaskService;
import com.nami.store.api.service.RemoteL2CacheService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import static com.nami.basic.constants.CommonConst.TASK_SEPARATOR;

@Slf4j
public class TaskEventHandler implements EventHandler<TaskEvent> {

    /**
     * 调优参数，调节数量
     */
    private static final int BATCH_SIZE = 50;

    private final RemoteConverTaskService taskService;

    private final RemoteL2CacheService l2CacheService;

    private final RedissonClient redisson;

    private final List<ConverTaskAddReq> batch = new CopyOnWriteArrayList<>();

    public TaskEventHandler(RemoteConverTaskService taskService, RemoteL2CacheService l2CacheService,
                            RedissonClient redisson) {
        this.taskService = taskService;
        this.l2CacheService = l2CacheService;
        this.redisson = redisson;
    }

    @Override
    public void onEvent(TaskEvent event, long sequence, boolean endOfBatch) {
        String taskKey = event.getTaskKey();
        String[] split = taskKey.split(TASK_SEPARATOR);
        String appKey = split[0];
        // 是否加入一级缓存 ， 混合使用，更快。如果有了，存入一级缓存，
        // 判断是否已经处理过
        if (l2CacheService.isExistByKey(ConverTaskUtil.buildConverItemTaskExistKey(appKey, taskKey))) {
            return; // 如果已经处理过，直接返回
        }

        RLock lock = redisson.getLock(buildConverItemTaskLockKey(appKey, taskKey));
        try {
            lock.lock();
            if (l2CacheService.isExistByKey(ConverTaskUtil.buildConverItemTaskExistKey(appKey, taskKey))) {
                return; // 如果已经处理过，直接返回
            }
            // 添加到批处理列表
            ConverTaskAddReq request = new ConverTaskAddReq();
            request.setTaskKey(taskKey);
            batch.add(request);
            l2CacheService.set(ConverTaskUtil.buildConverItemTaskExistKey(appKey, taskKey), 1);

            // 如果达到批处理大小，执行批量插入
            if (batch.size() >= BATCH_SIZE || endOfBatch) {
                processBatch();
            }
        } catch (Exception var3) {
            log.error("[disruptor] 出现错误 error:", var3);
        } finally {
            lock.unlock();
        }

    }

    private String buildConverItemTaskLockKey(String appKey, String taskKey) {
        return "nami:" + appKey + ":conver:lock:syncDb:" + taskKey;
    }

    private void processBatch() {
        if (batch.isEmpty()) {
            return;
        }
        // 批量插入数据库
        taskService.batchSave(new ArrayList<>(batch));
        log.info("[disruptor] 发送本批次数据：{}", batch.size());

        // 清空批处理列表和缓存
        batch.clear();
    }

}
